#include "sfu_client.hpp"
#include "p2pms_log.hpp"
#include "p2pms_util.hpp"
#include "capturer_track_source.hpp"

#include "mediasoupclient.hpp"

#include "rtc_base/logging.h"
#include "rtc_base/ssl_adapter.h"
#include "api/create_peerconnection_factory.h"
#include "api/audio_codecs/builtin_audio_decoder_factory.h"
#include "api/audio_codecs/builtin_audio_encoder_factory.h"
#include "api/video_codecs/video_decoder_factory_template.h"
#include "api/video_codecs/video_decoder_factory_template_dav1d_adapter.h"
#include "api/video_codecs/video_decoder_factory_template_libvpx_vp8_adapter.h"
#include "api/video_codecs/video_decoder_factory_template_libvpx_vp9_adapter.h"
#include "api/video_codecs/video_decoder_factory_template_open_h264_adapter.h"
#include "api/video_codecs/video_encoder_factory_template.h"
#include "api/video_codecs/video_encoder_factory_template_libaom_av1_adapter.h"
#include "api/video_codecs/video_encoder_factory_template_libvpx_vp8_adapter.h"
#include "api/video_codecs/video_encoder_factory_template_libvpx_vp9_adapter.h"
#include "api/video_codecs/video_encoder_factory_template_open_h264_adapter.h"

namespace p2pms
{

//--------------------------------------------------------------------------------------------------
//
//--------------------------------------------------------------------------------------------------
SFUClient::SFUClient(const std::string& user_id, ISFUClientHandler* handler, IPcFactoryPtr factory,
	rtc::Thread* thread, std::shared_ptr<IMsgDispatcher> dispatcher)
	: m_user_id(user_id)
	, m_pc_factory(factory)
	, m_client_handler(handler)
	, m_sdk_thread(thread)
	, m_dispatcher(dispatcher)
	, m_coro_runner(thread)
{
	assert(m_pc_factory);
	assert(m_client_handler);
	assert(m_sdk_thread);
	assert(m_dispatcher);
}

//--------------------------------------------------------------------------------------------------
//
//--------------------------------------------------------------------------------------------------
SFUClient::~SFUClient()
{
	ClearUp();
}

//--------------------------------------------------------------------------------------------------
//
//--------------------------------------------------------------------------------------------------
ErrCode SFUClient::Init()
{
	mediasoupclient::Initialize();

	m_dispatcher->Register(SFU_PREPARE_ROOM_RES, 
		[this](const Message& msg, CBType cbt) { OnPrepareRoomRes(msg, cbt); });
	m_dispatcher->Register(SFU_JOIN_ROOM_RES, 
		[this](const Message& msg, CBType cbt) { OnJoinRoomRes(msg, cbt); });
	m_dispatcher->Register(SFU_CONNECT_TRANSPORT_RES, 
		[this](const Message& msg, CBType cbt) { OnConnectTransportRes(msg, cbt); });
	m_dispatcher->Register(SFU_START_PRODUCE_RES, 
		[this](const Message& msg, CBType cbt) { OnStartProduceRes(msg, cbt); });
	m_dispatcher->Register(SFU_START_PUBLISH_NOTIFY, 
		[this](const Message& msg, CBType cbt) { OnStartPublishNotify(msg, cbt); });
	m_dispatcher->Register(SFU_STOP_PUBLISH_NOTIFY,
		[this](const Message& msg, CBType cbt) { OnStopPublishNotify(msg, cbt); });
	m_dispatcher->Register(SFU_START_PUBLISH_RES,
		[this](const Message& msg, CBType cbt) { OnStartPublishRes(msg, cbt); });
	m_dispatcher->Register(SFU_START_SUBSCRIBE_RES, 
		[this](const Message& msg, CBType cbt) { OnStartSubscribeRes(msg, cbt); });
	m_dispatcher->Register(SFU_STOP_PUBLISH_RES, 
		[this](const Message& msg, CBType cbt) { OnStopPublishRes(msg, cbt); });
	m_dispatcher->Register(SFU_STOP_SUBSCRIBE_RES, 
		[this](const Message& msg, CBType cbt) { OnStopSubscribeRes(msg, cbt); });

	LOG_INF("SFUClient initialized");

	return ERR_OK;
}

//--------------------------------------------------------------------------------------------------
//
//--------------------------------------------------------------------------------------------------
ErrCode SFUClient::JoinRoom(const std::string& room_id)
{
	m_room_id = room_id;

	return PrepareRoom() ? ERR_OK : ERR_FAILED;
}

//--------------------------------------------------------------------------------------------------
//
//--------------------------------------------------------------------------------------------------
ErrCode SFUClient::LeaveRoom()
{
	auto result = m_dispatcher->SendMsgSync(
		MakeSFULeaveRoomReqMsg(m_room_id, m_user_id), SFU_LEAVE_ROOM_RES);
	if (result.cbt == CBType::Timeout) {
		LOG_ERR("Leave room timeout");
		return ERR_FAILED;
	} 
	else if (result.cbt == CBType::Error) {
		LOG_ERR("Send leave room request failed");
		return ERR_FAILED;
	}

	LOG_INF("Leave room successfully");

	return ERR_OK;
}

//--------------------------------------------------------------------------------------------------
//
//--------------------------------------------------------------------------------------------------
ErrCode SFUClient::StartPublish(const MediaInfo& media)
{
	if (!ValidateMediaInfo(media)) {
		LOG_ERR("Invalid media info for publishing");
		return ERR_FAILED;
	}

	if (!StartProducing(media)) {
		LOG_ERR("Start producing failed");
		return ERR_FAILED;
	}

	return ERR_OK;
}

//--------------------------------------------------------------------------------------------------
//
//--------------------------------------------------------------------------------------------------
ErrCode SFUClient::StopPublish(const MediaInfo& media)
{
	auto iter = m_producers.find(media.media_id);
	if (iter == m_producers.end()) {
		LOG_ERR("Cannot find producer");
		return ERR_FAILED;
	}

	auto producer_id = iter->second.producer->GetId();

	// Close local producer first
	iter->second.producer->Close();
	m_producers.erase(iter);

	m_coro_runner.Run([this, media=media, producer_id]() {
		auto result = m_dispatcher->SendMsgCoro(MakeSFUStopProduceReqMsg(
			m_room_id, m_user_id, media, producer_id), SFU_STOP_PRODUCE_RES);
		if (result.cbt == CBType::Timeout) {
			LOG_ERR("Stop produce timeout");
			m_client_handler->OnStopPublishResult(media, false);
			return;
		} else if (result.cbt == CBType::Error) {
			LOG_ERR("Send stop produce request failed");
			m_client_handler->OnStopPublishResult(media, false);
			return;
		} else {
			int code = 0;
			SFUStopProduceResData res_data;
			if (!ParseResponse(result.msg, res_data, code)) {
				LOG_ERR("Parse stop produce response failed");
				m_client_handler->OnStopPublishResult(media, false);
				return;
			}
			if (code != 0) {
				LOG_ERR("Stop produce response error: {}", code);
				m_client_handler->OnStopPublishResult(media, false);
				return;
			}
			LOG_INF("Stop produce successfully, media: {}", ToString(media));
		}

		result = m_dispatcher->SendMsgCoro(MakeSFUStopPublishReqMsg(
			m_room_id, m_user_id, media), SFU_STOP_PUBLISH_RES);
		if (result.cbt == CBType::Timeout) {
			LOG_ERR("Stop publish timeout");
			m_client_handler->OnStopPublishResult(media, false);
		} else if (result.cbt == CBType::Error) {
			LOG_ERR("Send stop publish request failed");
			m_client_handler->OnStopPublishResult(media, false);
		} else {
			int code = 0;
			SFUStopPublishResData res_data;
			if (!ParseResponse(result.msg, res_data, code)) {
				LOG_ERR("Parse stop publish response failed");
				m_client_handler->OnStopPublishResult(media, false);
				return;
			}
			if (code != 0) {
				LOG_ERR("Stop publish response error: {}", code);
				m_client_handler->OnStopPublishResult(media, false);
				return;
			}
			LOG_INF("Stop publish successfully, media: {}", ToString(media));
			m_client_handler->OnStopPublishResult(media, true);
		}
	});

	return ERR_OK;
}

//--------------------------------------------------------------------------------------------------
//
//--------------------------------------------------------------------------------------------------
ErrCode SFUClient::StartSubscribe(const std::string& user_id, const MediaInfo& media)
{
	if (!ValidateMediaInfo(media)) {
		LOG_ERR("Invalid media info for subscription");
		return ERR_FAILED;
	}

	auto iter = m_consumers.find(media.media_id);
	if (iter != m_consumers.end()) {
		LOG_ERR("Consumer already exists, media:{}", ToString(media));
		return ERR_FAILED;
	}

	ConsumerInfo info;
	info.user_id = user_id;
	info.media = media;

	m_consumers.insert(std::make_pair(media.media_id, info));

	// Send start subscribe request
	m_dispatcher->SendMsg(MakeSFUStartSubscribeReqMsg(
		m_room_id, m_user_id, user_id, media), SFU_START_SUBSCRIBE_RES);

	return ERR_OK;
}

//--------------------------------------------------------------------------------------------------
//
//--------------------------------------------------------------------------------------------------
ErrCode SFUClient::StopSubscribe(const std::string& user_id, const MediaInfo& media)
{
	m_dispatcher->SendMsg(MakeSFUStopSubscribeReqMsg(
		m_room_id, user_id, media), SFU_STOP_SUBSCRIBE_RES);

	return ERR_OK;
}

//--------------------------------------------------------------------------------------------------
//
//--------------------------------------------------------------------------------------------------
void SFUClient::OnTimer()
{
	//for (const auto& item : m_producers) {
	//	if (m_send_transport && item.second.producer) {
	//		auto stats = m_send_transport->OnGetStats(item.second.producer);
	//		LOG_INF("Producer stats: {}", stats.dump());
	//	}
	//}

	//for (const auto& item : m_consumers) {
	//	if (m_recv_transport && item.second.consumer) {
	//		auto stats = m_recv_transport->OnGetStats(item.second.consumer);
	//		LOG_INF("Consumer stats: {}", stats.dump());
	//	}
	//}
}

////////////////////////////////////////////////////////////////////////////////////////////////////

//--------------------------------------------------------------------------------------------------
//
//--------------------------------------------------------------------------------------------------
void SFUClient::ClearUp()
{
	LOG_INF("Clearing up SFU call agent state");

	// Close transports
	if (m_send_transport) {
		m_send_transport->Close();
		m_send_transport = nullptr;
	}
	if (m_recv_transport) {
		m_recv_transport->Close();
		m_recv_transport = nullptr;
	}

	// Clear producers and consumers
	//for (auto& producer : m_producers) {
	//	producer.second->Close();
	//}
	m_producers.clear();

	//for (auto& consumer : m_consumers) {
	//	consumer.second->Close();
	//}
	m_consumers.clear();

	m_dispatcher->Unregister(SFU_PREPARE_ROOM_RES);
	m_dispatcher->Unregister(SFU_JOIN_ROOM_RES);
	m_dispatcher->Unregister(SFU_CONNECT_TRANSPORT_RES);
	m_dispatcher->Unregister(SFU_START_PRODUCE_RES);
	m_dispatcher->Unregister(SFU_START_PUBLISH_RES);
	m_dispatcher->Unregister(SFU_START_SUBSCRIBE_RES);
	m_dispatcher->Unregister(SFU_STOP_PUBLISH_RES);
	m_dispatcher->Unregister(SFU_STOP_SUBSCRIBE_RES);
}

//--------------------------------------------------------------------------------------------------
//
//--------------------------------------------------------------------------------------------------
bool SFUClient::PrepareRoom()
{
	return m_dispatcher->SendMsg(MakeSFUPrepareRoomReqMsg(m_user_id, m_room_id),
		SFU_PREPARE_ROOM_RES);
}

//--------------------------------------------------------------------------------------------------
//
//--------------------------------------------------------------------------------------------------
bool SFUClient::DoJoinRoom()
{
	return m_dispatcher->SendMsg(MakeSFUJoinRoomReqMsg(
		m_room_id, m_user_id, m_device.GetRtpCapabilities(), {}), SFU_JOIN_ROOM_RES);
}

//--------------------------------------------------------------------------------------------------
//
//--------------------------------------------------------------------------------------------------
bool SFUClient::CreateTransport(bool producing, bool consuming)
{
	LOG_INF("Creating SFU transport: producing={}, consuming={}", producing, consuming);

	// 同步创建 transport，等待服务器响应
	auto msg = MakeSFUCreateTransportReqMsg(
		m_room_id, 
		m_user_id, 
		"webrtc", 
		false, 
		producing, 
		consuming, 
		m_device.GetSctpCapabilities());

	auto result = m_dispatcher->SendMsgSync(msg, SFU_CREATE_TRANSPORT_RES);
	if (result.cbt == CBType::Error) {
		LOG_ERR("Failed to send create transport request");
		return false;
	}
	else if (result.cbt == CBType::Timeout) {
		LOG_ERR("Receive transport response timeout");
		return false;
	}

	int code = 0;
	SFUCreateTransportResData res_data;
	if (!ParseResponse(result.msg, res_data, code)) {
		LOG_ERR("Parse create transport response failed");
		return false;
	}

	if (code != 0) {
		LOG_ERR("Create transport response error: {}", code);
		return false;
	}

	// 修正 ice_candidates 字段兼容性
	for (auto& candidate : res_data.ice_candidates) {
		if (candidate.contains("address")) {
			candidate["ip"] = candidate["address"];
			candidate.erase("address");
		}
	}

	if (producing && !consuming) {
		m_send_transport = m_device.CreateSendTransport(
			this,
			res_data.transport_id,
			res_data.ice_parameters,
			res_data.ice_candidates,
			res_data.dtls_parameters);
		if (!m_send_transport) {
			LOG_ERR("Failed to create send transport");
			return false;
		}
		LOG_INF("Created send transport with ID: {}", res_data.transport_id);
	} 
	else if (!producing && consuming) {
		m_recv_transport = m_device.CreateRecvTransport(
			this,
			res_data.transport_id,
			res_data.ice_parameters,
			res_data.ice_candidates,
			res_data.dtls_parameters);
		if (!m_recv_transport) {
			LOG_ERR("Failed to create receive transport");
			return false;
		}
		LOG_INF("Created receive transport with ID: {}", res_data.transport_id);
	} else {
		LOG_ERR("Invalid transport type: producing={}, consuming={}", producing, consuming);
		return false;
	}

	LOG_INF("SFU transport created successfully: producing={}, consuming={}", 
		producing, consuming);

	return true;
}

//--------------------------------------------------------------------------------------------------
// SFU Protocol Message Handlers
//--------------------------------------------------------------------------------------------------
void SFUClient::OnPrepareRoomRes(const Message& msg, CBType cbt)
{
	assert(m_sdk_thread);
	m_sdk_thread->PostTask([this, msg, cbt]() {
		LOG_INF("Received prepare room response");

		if (cbt == CBType::Timeout) {
			LOG_ERR("Receive prepare room response timeout");
			m_client_handler->OnJoinRoomResult(false);
		}
		else if (cbt == CBType::Normal) {
			int code = 0;
			SFUPrepareRoomResData res_data;
			if (!ParseResponse(msg, res_data, code)) {
				LOG_ERR("Parse prepare room response failed");
				m_client_handler->OnJoinRoomResult(false);
				return;
			}

			if (code != 0) {
				LOG_ERR("Prepare room response error: {}", code);
				m_client_handler->OnJoinRoomResult(false);
				return;
			}

			mediasoupclient::PeerConnection::Options options;
			options.factory = m_pc_factory.get();
			m_device.Load(res_data.rtp_caps, &options);

			m_dispatcher->SendMsg(MakeSFUJoinRoomReqMsg(
				m_room_id, m_user_id, m_device.GetRtpCapabilities(), {}), SFU_JOIN_ROOM_RES);
		}
	});
}

//--------------------------------------------------------------------------------------------------
//
//--------------------------------------------------------------------------------------------------
void SFUClient::OnJoinRoomRes(const Message& msg, CBType cbt)
{
	// TODO: Parse response and handle room join result
	LOG_INF("Received join room response");
	
	assert(m_sdk_thread);
	m_sdk_thread->PostTask([this, msg, cbt]() {
		LOG_INF("Received join room response");
		if (cbt == CBType::Timeout) {
			LOG_ERR("Receive join room response timeout");
			m_client_handler->OnJoinRoomResult(false);
		}
		else if (cbt == CBType::Normal) {
			int code = 0;
			SFUJoinRoomResData res_data;
			if (!ParseResponse(msg, res_data, code)) {
				LOG_ERR("Parse join room response failed");
				m_client_handler->OnJoinRoomResult(false);
				return;
			}

			if (code != 0) {
				LOG_ERR("Join room response error: {}", code);
				m_client_handler->OnJoinRoomResult(false);
				return;
			}

			if (!CreateTransport(true, false)) {
				LOG_ERR("Failed to create send transport");
				m_client_handler->OnJoinRoomResult(false);
				return;
			}

			if (!CreateTransport(false, true)) {
				LOG_ERR("Failed to create recv transport");
				m_client_handler->OnJoinRoomResult(false);
				return;
			}

			m_client_handler->OnJoinRoomResult(true);
		}
	});
}

//--------------------------------------------------------------------------------------------------
// 注意：mediasoup 的 OnProuce 回调目前正阻塞 sdk 线程，所以这里的消息处理不能够切换到 sdk 线程
//--------------------------------------------------------------------------------------------------
void SFUClient::OnConnectTransportRes(const Message& msg, CBType cbt)
{
	LOG_INF("Received connect transport response");

	auto set_exception = [&](const std::string& err) {
		if (m_connect_promise) {
			m_connect_promise->set_exception(std::make_exception_ptr(std::runtime_error(err)));
			m_connect_promise.reset();
		}
	};

	auto set_value = [&]() {
		if (m_connect_promise) {
			m_connect_promise->set_value();
			m_connect_promise.reset();
		}
	};

	if (!m_connect_promise) {
		LOG_ERR("No pending promise for connect transport response");
		return;
	}

	int code = 0;
	SFUConnectTransportResData res_data;
	if (!ParseResponse(msg, res_data, code)) {
		set_exception("Failed to parse response");
		return;
	}

	if (res_data.transport_id.empty()) {
		LOG_ERR("Connect transport response missing transport ID");
		set_exception("Missing tranport ID");
		return;
	}

	if (code != 0) {
		LOG_ERR("Connect transport response error");
		set_exception("Response error");
		return;
	}

	set_value();
}

//--------------------------------------------------------------------------------------------------
// 注意：mediasoup 的 OnProuce 回调目前正阻塞 sdk 线程，所以这里的消息处理不能够切换到 sdk 线程
//--------------------------------------------------------------------------------------------------
void SFUClient::OnStartProduceRes(const Message& msg, CBType cbt)
{
	LOG_INF("Received produce response");

	auto set_exception = [&](const std::string& err) {
		if (m_produce_promise) {
			m_produce_promise->set_exception(std::make_exception_ptr(std::runtime_error(err)));
			m_produce_promise.reset();
		}
	};

	auto set_value = [&](const std::string& val) {
		if (m_produce_promise) {
			m_produce_promise->set_value(val);
			m_produce_promise.reset();
		}
	};

	if (!m_produce_promise) {
		LOG_ERR("No pending produce promise for response");
		return;
	}

	if (cbt == CBType::Timeout) {
		LOG_ERR("Produce response timeout");
		set_exception("Produce response timeout");
		return;
	}

	int code = 0;
	SFUStartProduceResData res_data;
	if (!ParseResponse(msg, res_data, code)) {
		set_exception("Parse response failed");
		return;
	}

	if (code != 0) {
		LOG_ERR("Produce response error");
		set_exception("Produce response error");
		return;
	}

	if (res_data.producer_id.empty()) {
		LOG_ERR("Produce response missing producer ID");
		set_exception("missing producer_id");
		return;
	}

	// 通知 produce 成功
	set_value(res_data.producer_id);

	LOG_INF("Produced media with ID: {}", res_data.producer_id);
	
	// Produce 成功后再发布媒体
	// 注意：这里需要切换到 sdk 线程，因为重入 websocket 线程会导致死锁
	assert(m_sdk_thread);
	m_sdk_thread->PostTask([this, res_data]() {
		m_dispatcher->SendMsg(MakeSFUStartPublishReqMsg(
			m_room_id, m_user_id, res_data.media), SFU_START_PUBLISH_RES);
	});
}

//--------------------------------------------------------------------------------------------------
//
//--------------------------------------------------------------------------------------------------
void SFUClient::OnStartPublishResTimeout(const Message& msg)
{
	SFUStartPublishReqData req_data;
	if (!ParseRequest(msg, req_data)) {
		LOG_ERR("Failed to parse start publish request data");
		return;
	}

	m_client_handler->OnStartPublishResult(req_data.media, false);

	LOG_INF("Start publish response timeout, media: {}", req_data.media.media_id);
}

//--------------------------------------------------------------------------------------------------
//
//--------------------------------------------------------------------------------------------------
void SFUClient::OnStartPublishResNormal(const Message& msg)
{
	LOG_INF("Received start publish response");

	int code = 0;
	SFUStartPublishResData res_data;
	if (!ParseResponse(msg, res_data, code)) {
		LOG_ERR("Failed to parse start publish response data");
		return;
	}
	
	if (code != 0) {
		LOG_ERR("Start publish response error");
		m_client_handler->OnStartPublishResult(res_data.media, false);
		return;
	}

	m_client_handler->OnStartPublishResult(res_data.media, true);

	LOG_INF("Start publish response received successfully");
}

//--------------------------------------------------------------------------------------------------
//
//--------------------------------------------------------------------------------------------------
void SFUClient::OnStartPublishRes(const Message& msg, CBType cbt)
{
	assert(m_sdk_thread);
	m_sdk_thread->PostTask([this, msg, cbt]() {
		(cbt == CBType::Timeout) ? OnStartPublishResTimeout(msg) : OnStartPublishResNormal(msg);
	});
}

//--------------------------------------------------------------------------------------------------
//
//--------------------------------------------------------------------------------------------------
rtc::scoped_refptr<webrtc::MediaStreamTrackInterface> SFUClient::CreateTrackFromMediaInfo(
	const MediaInfo& media)
{
	// 当前只处理设备类型媒体
	if (media.src_type != "device") {
		LOG_ERR("Unsupported media source type: {}", media.src_type);
		return nullptr;
	}

	if (m_producers.find(media.media_id) != m_producers.end()) {
		LOG_ERR("Producer info already exists");
		return nullptr;
	}

	rtc::scoped_refptr<webrtc::MediaStreamTrackInterface> media_track;
	if (media.media_type == "audio") {
		LOG_INF("Creating audio track with default device.");

		ProducerInfo info;
		info.media = media;

		info.audio_source = m_pc_factory->CreateAudioSource(cricket::AudioOptions());
		if (!info.audio_source) {
			LOG_ERR("Failed to create audio source for device: {}", media.src_id);
			return nullptr;
		}

		m_producers.insert(std::make_pair(media.media_id, info));
		
		return m_pc_factory->CreateAudioTrack(media.media_id, info.audio_source.get());
	} 
	else if (media.media_type == "video") {
		LOG_INF("Creating video track with device_id: {}", media.src_id);

		ProducerInfo info;
		info.media = media;

		info.video_source = CapturerTrackSource::Create(640, 480, 30, media.src_id);
		if (!info.video_source) {
			LOG_ERR("Failed to create video track source for device: {}", media.src_id);
			return nullptr;
		}

		m_producers.insert(std::make_pair(media.media_id, info));

		return m_pc_factory->CreateVideoTrack(media.media_id, info.video_source.get());
	} else {
		LOG_ERR("Unsupported media type: {}", media.media_type);
		return nullptr;
	}
}

//--------------------------------------------------------------------------------------------------
// 
//--------------------------------------------------------------------------------------------------
bool SFUClient::StartProducing(const MediaInfo& media)
{
	if (!m_device.CanProduce(media.media_type)) {
		LOG_ERR("Device cannot produce media:{}", media.media_type);
		return false;
	}

	auto track = CreateTrackFromMediaInfo(media);
	if (!track) {
		LOG_ERR("Failed to create track from media info");
		return false;
	}

	m_producing_media = media;

	auto producer = m_send_transport->Produce(this, track.get(), nullptr, nullptr, nullptr);
	if (!producer) {
		LOG_ERR("Failed to produce track: {}", media.media_id);
		return false;
	}

	LOG_INF("Produced track with ID: {}", media.media_id);

	auto iter = m_producers.find(media.media_id);
	if (iter == m_producers.end()) {
		LOG_ERR("Cannot find producer info");
		return false;
	}
	iter->second.producer = producer;

	return true;
}

//--------------------------------------------------------------------------------------------------
//
//--------------------------------------------------------------------------------------------------
void SFUClient::SendStartProduceAck(const std::string& rid, int code, const std::string& msg)
{
	assert(m_dispatcher);

	//m_dispatcher->SendMsg(MakeSFUStartProduceAckMsg(rid, code, msg, m_room_id, m_user_id));

	LOG_INF("Sent start produce ack with rid: {}, code: {}, msg: {}", rid, code, msg);
}

//--------------------------------------------------------------------------------------------------
// 
//--------------------------------------------------------------------------------------------------
void SFUClient::OnStopPublishNotify(const Message& msg, CBType cbt)
{
	LOG_INF("Received stop publish notify");

	assert(m_sdk_thread);
	m_sdk_thread->PostTask([this, msg]() {
		SFUStartPublishNotifyData notify_data;
		if (!ParseRequest(msg, notify_data)) {
			LOG_ERR("Parse stop publish notify failed");
			return;
		}

		if (!ValidateMediaInfo(notify_data.media)) {
			LOG_ERR("Invalid media information in start publish notify");
			return;
		}

		m_dispatcher->SendMsg(MakeSFUStopPublishAckMsg(
			msg.rid, notify_data.room_id, m_user_id, notify_data.media));

		auto iter = m_consumers.find(notify_data.media.media_id);
		if (iter == m_consumers.end()) {
			LOG_WRN("Cannot find consumer on received stop publish notify");
			return;
		}

		LOG_INF("Close and remove consumer: {}", iter->first);

		iter->second.consumer->Close();
		m_consumers.erase(iter);

		auto result = m_dispatcher->SendMsgSync(MakeSFUStopSubscribeReqMsg(
			m_room_id, m_user_id, notify_data.media), SFU_STOP_SUBSCRIBE_RES);
		if (result.cbt == CBType::Timeout) {
			LOG_ERR("Receive stop subscribe response timeout");
		}
		else if (result.cbt == CBType::Error) {
			LOG_ERR("Send stop subscribe request failed");
		} 
		else if (result.cbt == CBType::Normal) {
			int code = 0;
			SFUStopSubscribeResData res_data;
			if (!ParseResponse(result.msg, res_data, code)) {
				LOG_ERR("Parse stop subscribe response failed");
			}
			else {
				if (code != 0) {
					LOG_ERR("Stop subscribe response error: {}", code);
				}
				else {
					LOG_INF("Stop subscribe response success");
				}
			}
		}

		// 无论停止订阅是否成功，都通知客户端处理停止发布事件
		LOG_INF("Notifying client handler about stop publish notify");
		m_client_handler->OnStopPublishNotify(notify_data.user_id, notify_data.media);
	});
}

//--------------------------------------------------------------------------------------------------
// 
//--------------------------------------------------------------------------------------------------
void SFUClient::OnStartPublishNotify(const Message& msg, CBType cbt)
{
	LOG_INF("Received start publish notify");

	assert(m_sdk_thread);
	m_sdk_thread->PostTask([this, msg]() {
		SFUStartPublishNotifyData notify_data;
		if (!ParseRequest(msg, notify_data)) {
			LOG_ERR("Parse start publish notify failed");
			return;
		}

		if (!ValidateMediaInfo(notify_data.media)) {
			LOG_ERR("Invalid media information in start publish notify");
			return;
		}

		m_dispatcher->SendMsg(MakeSFUStartPublishAckMsg(
			msg.rid, notify_data.room_id, m_user_id, notify_data.media));

		m_client_handler->OnStartPublishNotify(notify_data.user_id, notify_data.media);
	});
}

//--------------------------------------------------------------------------------------------------
// 
//--------------------------------------------------------------------------------------------------
void SFUClient::OnStartSubscribeResTimeout(const Message& msg)
{
	LOG_ERR("Start subscribe response timeout");

	SFUStartSubscribeReqData req_data;
	if (!ParseRequest(msg, req_data)) {
		LOG_ERR("Failed to parse start subscribe request");
		return;
	}

	auto iter = m_consumers.find(req_data.media.media_id);
	if (iter == m_consumers.end()) {
		LOG_ERR("Consumer does not exist");
		return;
	}

	m_consumers.erase(iter);
	m_client_handler->OnStartSubscribeResult(req_data.user_id, req_data.media, false);
}

//--------------------------------------------------------------------------------------------------
// 
//--------------------------------------------------------------------------------------------------
void SFUClient::OnStartSubscribeResNormal(const Message& msg)
{
	LOG_INF("Start subscribe response received");

	int code = 0;
	SFUStartSubscribeResData res_data;
	if (!ParseResponse(msg, res_data, code)) {
		LOG_ERR("Failed to parse start subscribe response");
		return;
	}

	auto iter = m_consumers.find(res_data.media.media_id);
	if (iter == m_consumers.end()) {
		LOG_ERR("Consumer does not exist on received start subscribe response");
		return;
	}

	if (code != 0) {
		LOG_ERR("Start subscribe response error");
		m_consumers.erase(iter);
		m_client_handler->OnStartSubscribeResult(res_data.user_id, res_data.media, false);
		return;
	}

	auto consumer = m_recv_transport->Consume(this, 
		res_data.consumer_id, 
		res_data.producer_id, 
		res_data.media.media_type, 
		&res_data.rtp_parameters, 
		res_data.app_data);
	if (!consumer) {
		LOG_ERR("Failed to create consumer for media: {}", res_data.media.media_id);
		m_consumers.erase(iter);
		m_client_handler->OnStartSubscribeResult(res_data.user_id, res_data.media, false);
		return;
	}

	iter->second.consumer = consumer;
	iter->second.producer_id = res_data.producer_id;

	m_client_handler->OnAddTrack(res_data.pub_user_id, res_data.media, consumer->GetTrack());

	LOG_INF("Successfully started subscribing to media with consumer ID: {}", 
		res_data.consumer_id);

	m_client_handler->OnStartSubscribeResult(res_data.user_id, res_data.media, true);
}

//--------------------------------------------------------------------------------------------------
// 
//--------------------------------------------------------------------------------------------------
void SFUClient::OnStartSubscribeRes(const Message& msg, CBType cbt)
{
	assert(m_sdk_thread);
	m_sdk_thread->PostTask([this, msg, cbt]() {
		if (cbt == CBType::Timeout) {
			OnStartSubscribeResTimeout(msg);
		} else {
			OnStartSubscribeResNormal(msg);
		}
	});
}

//--------------------------------------------------------------------------------------------------
// 
//--------------------------------------------------------------------------------------------------
void SFUClient::OnStopPublishResTimeout(const Message& msg)
{
	LOG_ERR("Stop publish response timeout");

	SFUStopPublishReqData req_data;
	if (!ParseRequest(msg, req_data)) {
		LOG_ERR("Failed to parse stop publish request");
		return;
	}

	auto it = m_producers.find(req_data.media.media_id);
	if (it != m_producers.end()) {
		if (it->second.producer) {
			it->second.producer->Close();
		}
		m_producers.erase(it);
	}

	m_client_handler->OnStopPublishResult(req_data.media, false);
}

//--------------------------------------------------------------------------------------------------
// 
//--------------------------------------------------------------------------------------------------
void SFUClient::OnStopPublishResNormal(const Message& msg)
{
	LOG_INF("Stop publish response received");

	int code = 0;
	SFUStopPublishResData res_data;
	if (!ParseResponse(msg, res_data, code)) {
		LOG_ERR("Failed to parse stop publish response");
		return;
	}

	if (code != 0) {
		LOG_ERR("Stop publish response error");
		m_client_handler->OnStopPublishResult(res_data.media, false);
		return;
	}

	auto it = m_producers.find(res_data.media.media_id);
	if (it != m_producers.end()) {
		if (it->second.producer) {
			it->second.producer->Close();
		}
		m_producers.erase(it);
	}

	m_client_handler->OnStopPublishResult(res_data.media, true);

	LOG_INF("Successfully stopped publishing media: {}", res_data.media.media_id);
}

//--------------------------------------------------------------------------------------------------
// 
//--------------------------------------------------------------------------------------------------
void SFUClient::OnStopPublishRes(const Message& msg, CBType cbt)
{
	assert(m_sdk_thread);
	m_sdk_thread->PostTask([this, msg, cbt]() {
		(cbt == CBType::Timeout) ? OnStopPublishResTimeout(msg) : OnStopPublishResNormal(msg);
	});
}

//--------------------------------------------------------------------------------------------------
// 
//--------------------------------------------------------------------------------------------------
void SFUClient::OnStopSubscribeResTimeout(const Message& msg)
{
	LOG_ERR("Stop subscribe response timeout");

	SFUStopSubscribeReqData req_data;
	if (!ParseRequest(msg, req_data)) {
		LOG_ERR("Failed to parse stop subscribe request");
		return;
	}

	auto iter = m_consumers.find(req_data.media.media_id);
	if (iter == m_consumers.end()) {
		LOG_ERR("Consumer does not exist");
		return;
	}

	m_client_handler->OnStopSubscribeResult(req_data.user_id, req_data.media, false);
}

//--------------------------------------------------------------------------------------------------
// 
//--------------------------------------------------------------------------------------------------
void SFUClient::OnStopSubscribeResNormal(const Message& msg)
{
	LOG_INF("Stop subscribe response received");

	int code = 0;
	SFUStopSubscribeResData res_data;
	if (!ParseResponse(msg, res_data, code)) {
		LOG_ERR("Failed to parse stop subscribe response");
		return;
	}

	auto iter = m_consumers.find(res_data.media.media_id);
	if (iter == m_consumers.end()) {
		LOG_ERR("Consumer does not exist on received start subscribe response");
		return;
	}

	if (code != 0) {
		LOG_ERR("Stop subscribe response error");
		m_client_handler->OnStopSubscribeResult(res_data.user_id, res_data.media, false);
		return;
	}

	m_consumers.erase(iter);
	m_client_handler->OnStopSubscribeResult(res_data.user_id, res_data.media, true);

	LOG_INF("Successfully stopped subscribing to media: {}", res_data.media.media_id);
}

//--------------------------------------------------------------------------------------------------
// 
//--------------------------------------------------------------------------------------------------
void SFUClient::OnStopSubscribeRes(const Message& msg, CBType cbt)
{
	assert(m_sdk_thread);
	m_sdk_thread->PostTask([this, msg, cbt]() {
		(cbt == CBType::Timeout) ? OnStopSubscribeResTimeout(msg) : OnStopSubscribeResNormal(msg);
	});
}

//--------------------------------------------------------------------------------------------------
// Call from mediasoupclient::ProducerListener
//--------------------------------------------------------------------------------------------------
std::future<std::string> SFUClient::OnProduce(mediasoupclient::SendTransport* transport,
	const std::string& kind, nlohmann::json rtpParameters, const nlohmann::json& appData)
{
	LOG_INF("OnProduce called with kind: {}, rtpParameters: {}", kind, rtpParameters.dump());

	if (transport->GetId() != m_send_transport->GetId()) {
		LOG_ERR("Unknown transport requested to produce: {}", transport->GetId());
		std::promise<std::string> promise;
		promise.set_exception(std::make_exception_ptr(std::runtime_error("Unknown transport")));
		return promise.get_future();
	}

	// 等待服务器响应
	m_produce_promise.reset(new std::promise<std::string>());
	auto future = m_produce_promise->get_future();

	m_dispatcher->SendMsg(MakeSFUStartProduceReqMsg(m_room_id, m_user_id, m_producing_media, 
		transport->GetId(), kind, rtpParameters, appData), SFU_START_PRODUCE_RES);

	return future;
}

//--------------------------------------------------------------------------------------------------
// Call from mediasoupclient::ProducerListener
//--------------------------------------------------------------------------------------------------
std::future<std::string> SFUClient::OnProduceData(mediasoupclient::SendTransport* transport,
	const nlohmann::json& sctpStreamParameters, const std::string& label,
	const std::string& protocol, const nlohmann::json& appData)
{
	LOG_INF("OnProduceData called with label: {}, protocol: {}", label, protocol);
}

//--------------------------------------------------------------------------------------------------
// 注意：没有切换到 sdk 线程执行，因为这个函数可能在 mediasoupclient 的回调中被调用
//--------------------------------------------------------------------------------------------------
std::future<void> SFUClient::OnConnect(mediasoupclient::Transport* transport,
	const nlohmann::json& dtlsParameters)
{
	LOG_INF("OnConnect called for transport: {}", transport->GetId());

	const std::string transport_id = transport->GetId();

	std::string send_or_recv;
	if (transport_id == m_send_transport->GetId()) {
		send_or_recv = "send";
	}
	else if (transport_id == m_recv_transport->GetId()) {
		send_or_recv = "recv";
	}
	else {
		std::promise<void> promise;
		promise.set_exception(std::make_exception_ptr("Unknown transport requested to connect"));
		return promise.get_future();
	}

	// 等待服务器响应
	m_connect_promise.reset(new std::promise<void>());
	auto future = m_connect_promise->get_future();

	m_dispatcher->SendMsg(MakeSFUConnectTransportReqMsg(m_room_id, m_user_id, "webrtc", 
		transport_id, send_or_recv, dtlsParameters), SFU_CONNECT_TRANSPORT_RES);

	LOG_INF("Sent connect transport request for transport: {}", transport_id);

	return future;
}

//--------------------------------------------------------------------------------------------------
// 
//--------------------------------------------------------------------------------------------------
void SFUClient::OnConnectionStateChange(mediasoupclient::Transport* transport,
	const std::string& connectionState)
{
	LOG_INF("Transport connection state changed: transportId={}, state={}", 
		transport->GetId(), connectionState);
}

//--------------------------------------------------------------------------------------------------
// 
//--------------------------------------------------------------------------------------------------
void SFUClient::OnTransportClose(mediasoupclient::Producer* producer)
{
	LOG_INF("Producer transport closed: producerId={}", producer->GetId());
}

//--------------------------------------------------------------------------------------------------
// 
//--------------------------------------------------------------------------------------------------
void SFUClient::OnTransportClose(mediasoupclient::Consumer* consumer)
{
	LOG_INF("Consumer transport closed: consumerId={}", consumer->GetId());
}

}
